package com.dy.realtime.dwd;

import com.dy.realtime.common.base.BaseSqlApp;
import com.dy.realtime.common.constant.Constant;
import com.dy.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @projectName: MedicalTreatment
 * @package: com.dy.realtime.dwd
 * @className: DwdUserPatientAdd
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/12 11:30
 * @version: 1.0
 */

public class DwdUserPatientAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdUserPatientAdd().start(Constant.TOPIC_DWD_TRADE_USER_PATIENT_ADD,Constant.TOPIC_DWD_TRADE_USER_PATIENT_ADD,4,10017);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
       readOdsDb(tableEnv,groupId);
        Table patient = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id ,\n" +
                "    `data`['create_time'] create_time ,\n" +
                "    `data`['update_time'] update_time ,\n" +
                "    `data`['birthday'] birthday ,\n" +
                "    `data`['gender'] gender ,\n" +
                "    concat(substring(`data`['name'],1,1),'**') name ,\n" +
                "    `data`['user_id'] user_id \n" +
                "from topic_db_medical\n" +
                "where `table` = 'patient' and `type` ='insert'");
//        patient.execute().print();
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_USER_PATIENT_ADD+"(\n" +
                "    id string,\n" +
                "    create_time string,\n" +
                "    update_time string,\n" +
                "    birthday string,\n" +
                "    gender string,\n" +
                "    name string,\n" +
                "    user_id string\n" +
                ")"+ SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_USER_PATIENT_ADD));
         patient.executeInsert(Constant.TOPIC_DWD_TRADE_USER_PATIENT_ADD);
    }
}
